package top.javatool.canal.client.client;

import com.alibaba.otter.canal.client.rabbitmq.RabbitMQCanalConnector;
import com.alibaba.otter.canal.protocol.FlatMessage;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import top.javatool.canal.client.handler.MessageHandler;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author Mr.Zihan
 * @date 2021年10月13日15:13
 */

public class RabbitMqCanalClient extends AbstractCanalClient {

    private Logger log = LoggerFactory.getLogger(RabbitMqCanalClient.class);


    public static Builder builder() {
        return Builder.builder();
    }


    @Override
    public void process() {
        RabbitMQCanalConnector connector = (RabbitMQCanalConnector) getConnector();
        MessageHandler messageHandler = getMessageHandler();
        while (flag) {
            try {
                connector.connect();
                connector.subscribe();
                while (flag) {
                    try {
                        List<FlatMessage> messages = connector.getFlatListWithoutAck(timeout, unit);
                        log.info("获取消息 {}", messages);
                        if (messages != null) {
                            for (FlatMessage flatMessage : messages) {
                                messageHandler.handleMessage(flatMessage);
                            }
                        }
                        connector.ack();
                    } catch (Exception e) {
                        log.error("canal 消费异常", e);
                    }
                }
            } catch (Exception e) {
                log.error("canal 连接异常", e);
            }
        }
        connector.unsubscribe();
        connector.disconnect();
    }


    public static final class Builder {
        private String filter = StringUtils.EMPTY;
        private Integer batchSize = 1;
        private Long timeout = 1L;
        private TimeUnit unit = TimeUnit.SECONDS;
        private String servers;
        private String queueName;
        private String vhost;
        private String username;
        private String password;
        private MessageHandler messageHandler;

        private Builder() {
        }

        public static Builder builder() {
            return new Builder();
        }

        public Builder servers(String servers) {
            this.servers = servers;
            return this;
        }

        public Builder topic(String topic) {
            this.queueName = topic;
            return this;
        }

        public Builder vhost(String vhost) {
            this.vhost = vhost;
            return this;
        }
        public Builder username(String username) {
            this.username = username;
            return this;
        }
        public Builder password(String password) {
            this.password = password;
            return this;
        }

        public Builder filter(String filter) {
            this.filter = filter;
            return this;
        }

        public Builder batchSize(Integer batchSize) {
            this.batchSize = batchSize;
            return this;
        }

        public Builder timeout(Long timeout) {
            this.timeout = timeout;
            return this;
        }

        public Builder unit(TimeUnit unit) {
            this.unit = unit;
            return this;
        }

        public Builder messageHandler(MessageHandler messageHandler) {
            this.messageHandler = messageHandler;
            return this;
        }

        public RabbitMqCanalClient build() {
            RabbitMQCanalConnector connector = new RabbitMQCanalConnector(servers, vhost, queueName, null, null, username, password, null, true);
            RabbitMqCanalClient rabbitMqCanalClient = new RabbitMqCanalClient();
            rabbitMqCanalClient.setMessageHandler(messageHandler);
            rabbitMqCanalClient.setConnector(connector);
            rabbitMqCanalClient.filter = this.filter;
            rabbitMqCanalClient.unit = this.unit;
            rabbitMqCanalClient.batchSize = this.batchSize;
            rabbitMqCanalClient.timeout = this.timeout;
            return rabbitMqCanalClient;
        }
    }
}
